package com.atguigu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternFlatSelectFunction;
import org.apache.flink.cep.PatternFlatTimeoutFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.util.List;
import java.util.Map;

/*
访客跳出情况判断
 */
public class UserJumpDetailApp {
    public static void main(String[] args) throws Exception {
        //TODO 0. 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        //TODO 0.0 设定事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        //TODO 1. 从Kafka的dwd_page_log主题中读取页面日志
        String sourceTopic = "dwd_page_log";
        String groupId = "userJumpDetailApp";
        String sinkTopic = "dwm_user_jump_detail";

        DataStreamSource<String> dataStreamSource = env.addSource(
                MyKafkaUtil.getKafkaSource(sourceTopic,groupId)
        );

        DataStream<JSONObject> jsonObjectDataStream = dataStreamSource.map(JSON::parseObject);
        jsonObjectDataStream.print("json::");

        // TODO 2. 指定事件时间字段
        /*
        设定事件时间戳和水位线，
        设定递增的水印 和 时间戳；

        如果时间是事件时间，只需要指定事件时间的时间戳字段即可，以及水印，以确保 乱序数据的而有效处理；

        因为在10秒钟以内判断，需要判断该日志的事件时间；


         */
        SingleOutputStreamOperator<JSONObject> jsonObjectWithEtDS =
                jsonObjectDataStream.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forMonotonousTimestamps().withTimestampAssigner(
                        new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject jsonObject, long l) {
                                return jsonObject.getLong("ts");
                            }
                        }
                ));

        // TODO 3. 根据日志数据的mid进行分组
        KeyedStream<JSONObject,String> jsonObjectStringKeyedStream =
                jsonObjectWithEtDS.keyBy(jsonObj -> jsonObj.getJSONObject("common").getString("mid"));

        //TODO 4. 配置CEP表达式
        Pattern<JSONObject,JSONObject> pattern = Pattern.<JSONObject>begin("GoIn").where(
                new SimpleCondition<JSONObject>() {
                    //条件1 ， 进入的第一个页面
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");
                        System.out.println("first in: "+ lastPageId);
                        if(lastPageId == null || lastPageId.length() ==0){
                            return true;
                        }
                        return false;
                    }
                }
        ).next("next").where(
                new SimpleCondition<JSONObject>() {
                    //条件2 在10秒时间范围内必须有第二个页面
                    @Override
                    public boolean filter(JSONObject jsonObject) throws Exception {
                        String pageId = jsonObject.getJSONObject("page").getString("page_id");
                        System.out.println("next: "+ pageId);
                        if(pageId == null || pageId.length() ==0){
                            return true;
                        }
                        return false;
                    }
                }
                //定义在10秒内的动作
        ).within(Time.milliseconds(10000));

        //TODO 5. 根据表达式筛选流
        PatternStream<JSONObject> patternStream = CEP.pattern(jsonObjectStringKeyedStream,pattern);

        // TODO 6.提取命中的数据
        /*
        Pattern 和select是组合使用的
        flatSelect 方法: 传入一个侧输出流tag， 和一个patternFlatTimeoutFunction 函数
            提取满足Pattern的数据； pattern 是满足的数据，然后将数据装入output

         */

        final OutputTag<String> timeoutTag  = new OutputTag<String>("timeout"){};

        SingleOutputStreamOperator<String> filteredDS = patternStream.flatSelect(timeoutTag,
                new PatternFlatTimeoutFunction<JSONObject, String>() {
             //没有超时的数据不在统计范围之内，超时的数据，说明不满足上面的模式匹配，
                    @Override
                    public void timeout(Map<String, List<JSONObject>> pattern, long l, Collector<String> collector) throws Exception {
                        /*
                        参数 pattern是上面的first 和next 的数据的集合

                         */
                        List<JSONObject> objectList = pattern.get("GoIn");
                        //这里进入out的数据都被timeoutTag标记
                        for (JSONObject jsonObject : objectList) {
                            collector.collect(jsonObject.toJSONString());
                        }
                    }
                },
                new PatternFlatSelectFunction<JSONObject, String>() {
                    @Override
                    public void flatSelect(Map<String, List<JSONObject>> pattern, Collector<String> collector) throws Exception {
                        // 因为不超时的事件不提取，所以这里不写代码
                    }
                }
        );

        // 通过SideOutput侧输出流输出超时数据
        DataStream<String> jumpDS = filteredDS.getSideOutput(timeoutTag);
        jumpDS.print("jump::: ");


        //TODO 7. 将跳出数据写回到kafka的DWM层
        jumpDS.addSink(MyKafkaUtil.getKafkaSink(sinkTopic));


        env.execute();
    }
}
